package ranks.tasks.count_seeds;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.lib.NullOutputFormat;

public class CountSeeds extends Job {

	public CountSeeds(String... settings) {
		super("Count Seeds", settings);

		this.setConfig("-reduce", 0);
	}

	@Override
	protected CombineConfig getCombineConfig() {
		return CombineConfig.newInstance(null);
	}

	@Override
	protected InputConfig getInputConfig(FileSystem fs) throws IOException {
		return InputConfig.newInstance(KeyValueTextInputFormat.class, this
				.getConfig("-linkgraph"));
	}

	@Override
	protected MapConfig getMapConfig() {
		return MapConfig.newInstance(CountSeeds_Mapper.class, 0, Text.class,
				Text.class, null);
	}

	@Override
	protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
		return OutputConfig.newInstance(NullOutputFormat.class, Text.class,
				Text.class, this.getConfig("-output") + "/CountSeeds", null);
	}

	@Override
	protected ReduceConfig getReduceConfig() {
		return ReduceConfig.newInstance(null, 0);
	}

	@Override
	protected void setUp(JobConf conf, FileSystem fs) throws IOException {
		this.setJobConfConfig("Seeds", this.getConfig("-seeds"));
	}

	@Override
	protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
			throws IOException {
		long websize = this.getCounter(job, "WebSize", "Number");
		long seeds = this.getCounter(job, "Seeds", "Number");

		double times = 1.0 * websize / seeds;

		BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs
				.create(new Path(this.getConfig("-seeds") + ".times"), true)));
		bw.write(String.valueOf(times));
		bw.newLine();
		bw.flush();
		bw.close();
		
		System.out.println("Times = " + times);
	}

}
